package com.project.pulsar.delayMsgAndSubscriptionsModel;

import com.project.pulsar.conf.PulsarConf;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * Key共享模式
 */
@RestController
public class KeyShareMessageController {
    @Autowired
    PulsarConf pulsarConf;

    String tenantName = "delay";//调用接口创建的租户,创建方法见com.project.pulsar.persistent.PersistentController.java
    String namespace = "mySpace";//调用接口创建的命名空间,创建方法见com.project.pulsar.persistent.PersistentController.java

    /**
     * 生产消息
     *
     * @param msg
     * @throws PulsarClientException
     */
    @GetMapping("/keyShare/sendMsg")
    public MessageId sendMsg(String msg) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        String topic = "persistent://" + tenantName + "/" + namespace + "/my-topic";
        //持久化         租户   命名空间   主题
        Producer<byte[]> producer1 = pulsarFactory.newProducer()
                .topic(topic)
                .create();

        return producer1.send(msg.getBytes());
    }



    @Bean
    public void comsumerByListener1() throws PulsarClientException {
        String topic = "persistent://" + tenantName + "/" + namespace + "/my-topic";
        MessageListener myMessageListener = (consumer, msg) -> {
            try {
                System.out.println("comsumerByListener1: " + new String(msg.getData()));
                consumer.acknowledge(msg);
            } catch (Exception e) {
                consumer.negativeAcknowledge(msg);
            }
        };
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        pulsarFactory.newConsumer()
                .topic(topic)
                .subscriptionName("my-subscriptionByListener")//必须名称一致，才可触发下一行，否则就是两个不同的消费者
                .subscriptionType(SubscriptionType.Key_Shared)
                .messageListener(myMessageListener)
                .subscribe();
    }

    @Bean
    public void comsumerByListener2() throws PulsarClientException {
        String topic = "persistent://" + tenantName + "/" + namespace + "/my-topic";
        MessageListener myMessageListener = (consumer, msg) -> {
            try {
                System.out.println("comsumerByListener2: " + new String(msg.getData()));
                consumer.acknowledge(msg);
            } catch (Exception e) {
                consumer.negativeAcknowledge(msg);
            }
        };
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        pulsarFactory.newConsumer()
                .topic(topic)
                .subscriptionName("my-subscriptionByListener")//必须名称一致，才可触发下一行，否则就是两个不同的消费者
                .subscriptionType(SubscriptionType.Key_Shared)
                .messageListener(myMessageListener)
                .subscribe();
    }
}